package com.xunyin.nettyV;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

public class BootNettyMqttMsgBack {


    private static final Logger log = LoggerFactory.getLogger(BootNettyMqttMsgBack.class);

    /**
     * 确认连接请求
     *
     * @param channel
     * @param mqttMessage
     */
    public static void connack(Channel channel, MqttMessage mqttMessage) {
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();
        MqttConnectPayload mqttConnectPayload = (MqttConnectPayload) mqttMessage.payload();

        //log.info(" mqttConnectPayload.clientIdentifier():"+mqttConnectPayload.clientIdentifier());
        String clientId = mqttConnectPayload.clientIdentifier();

        //	构建返回报文， 可变报头
        MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeaderInfo.isCleanSession());
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_LEAST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
        //	构建CONNACK消息体
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);

        log.info("back--" + connAck.toString());

        log.debug("设备上线,channelId：{}，clientId：{}", channel.id(), clientId);

        channel.writeAndFlush(connAck);

        BootNettyMqttServer.MQTTdeviceChannelGroup.add(channel);
    }

    public static void disconnack(Channel channel, MqttMessage mqttMessage) {
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();

        MqttConnectPayload mqttConnectPayload = (MqttConnectPayload) mqttMessage.payload();

        String clientId = mqttConnectPayload.clientIdentifier();

        //	构建返回报文， 可变报头
        MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BANNED, mqttConnectVariableHeaderInfo.isCleanSession());
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.DISCONNECT, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
        //	构建CONNACK消息体
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
        //log.info("back--"+connAck.toString());

        log.debug("设备下线,channelId：{}，clientId：{}", channel.id(), clientId);

        channel.writeAndFlush(connAck);

        //MQTTdeviceRemove(channel);
        BootNettyMqttServer.MQTTdeviceChannelGroup.remove(channel);
    }

    /**
     * 根据qos发布确认
     *
     * @param channel
     * @param mqttMessage
     */
    public static void puback(Channel channel, MqttMessage mqttMessage) {
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
        MqttQoS qos = (MqttQoS) mqttFixedHeaderInfo.qosLevel();
        byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
        mqttPublishMessage.payload().readBytes(headBytes);
        String data = new String(headBytes);
        System.out.println("publish data--" + data);

        String payloadStr = data;
        JSONObject jsonPayloadObject = JSON.parseObject(payloadStr);

        // 硬件物理地址
        //String mac = jsonPayloadObject.getString("mac");
        //System.out.println("硬件-mac:"+mac);

        String topicName1 = mqttPublishMessage.variableHeader().topicName();
        log.info("接收到topicName1：" + topicName1);

        byte[] bytes = data.getBytes(CharsetUtil.UTF_8);
        ByteBuf buf = Unpooled.wrappedBuffer(bytes);

        switch (qos) {
            case AT_MOST_ONCE:        //	至多一次
                System.out.println("AT_MOST_ONCE");
/*                //	构建返回报文， 可变报头
                MqttPublishVariableHeader mqttMessageIdVariableHeaderBack0 =new MqttPublishVariableHeader(mqttPublishMessage.variableHeader().topicName(),mqttPublishMessage.variableHeader().packetId());
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack0 = new MqttFixedHeader(MqttMessageType.PUBLISH, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                //	构建publish消息体
                MqttPublishMessage publishMessage0 = new MqttPublishMessage(mqttFixedHeaderBack0, mqttMessageIdVariableHeaderBack0,
                        buf);
                channel.writeAndFlush(publishMessage0);*/
                //推送到订阅的客户端
                subscribSend(mqttMessage, buf);
                break;
            case AT_LEAST_ONCE:        //	至少一次
                System.out.println("AT_LEAST_ONCE");

                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                //	构建PUBACK消息体
                MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                log.info("back--" + pubAck.toString());
                channel.writeAndFlush(pubAck);

                //推送到订阅的客户端
                subscribSend(mqttMessage, buf);


                break;
            case EXACTLY_ONCE:        //	刚好一次
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2, mqttMessageIdVariableHeaderBack2);
                log.info("back--" + mqttMessageBack.toString());

                channel.writeAndFlush(mqttMessageBack);
                break;
            default:
                break;
        }
    }


    /**
     * 订阅推送
     */

    public static void subscribSend(MqttMessage mqttMessage, ByteBuf buf) {
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        Object obj = mqttMessage.variableHeader();
        MqttPublishVariableHeader variableHeader = (MqttPublishVariableHeader) obj;
        String topicName = variableHeader.topicName();
        int packetId = variableHeader.packetId();
        //固定消息头 注意此处的消息类型PUBLISH mqtt协议
        MqttFixedHeader FixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0);
        //可变消息头
        MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(topicName, packetId);
        //推送消息体

        MqttPublishMessage mqttPublishMessageResult = new MqttPublishMessage(FixedHeader, mqttPublishVariableHeader, buf);

        log.info("推送地址————》" + topicName);
        if (BootNettyMqttServer.subscribeMap.containsKey(topicName)) {
            List<ChannelId> channelList = BootNettyMqttServer.subscribeMap.get(topicName);
            for (int i = 0; i < channelList.size(); i++) {
                //订阅次此topic的Mqtt客户端搜到此消息，
                Channel channelSub = BootNettyMqttServer.MQTTdeviceChannelGroup.find(channelList.get(i));
                if (channelSub != null) {
                    //writeAndFlush会将ByteBuf的引用释放refCnt会减去1,使用retain加1
                    mqttPublishMessageResult.retain();
                    channelSub.writeAndFlush(mqttPublishMessageResult);
                }

            }
            mqttPublishMessageResult.release();

        }
    }


    /**
     * 发布完成 qos2
     *
     * @param channel
     * @param mqttMessage
     */
    public static void pubcomp(Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
        //log.info("back--"+mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }

    /**
     * 订阅确认
     *
     * @param channel
     * @param mqttMessage
     */
    public static void suback(Channel channel, MqttMessage mqttMessage) {
        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());
        //log.info(topics.toString());
        List<Integer> grantedQoSLevels = new ArrayList<>(topics.size());
        for (int i = 0; i < topics.size(); i++) {
            grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }
        //	构建返回报文	有效负载
        MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 2 + topics.size());
        //	构建返回报文	订阅确认
        MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
        log.info("back--" + subAck.toString());
        // BootNettyMqttServer.MQTTdeviceChannelGroup.add(channel);
        channel.writeAndFlush(subAck);
    }

    /**
     * 取消订阅确认
     *
     * @param channel
     * @param mqttMessage
     */
    public static void unsuback(Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文	可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 2);
        //	构建返回报文	取消订阅确认
        MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack, variableHeaderBack);
        log.info("back--" + unSubAck.toString());
        //disconnect(channel,mqttMessage);
        channel.writeAndFlush(unSubAck);
    }

    /**
     * 心跳响应
     *
     * @param channel
     * @param mqttMessage
     */
    public static void pingresp(Channel channel, MqttMessage mqttMessage) {
        //	心跳响应报文	11010000 00000000  固定报文
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_LEAST_ONCE, false, 0);
        MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
        log.info("back--" + mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }

    public static void disconnect(Channel channel, MqttMessage mqttMessage) {
        //channelMqttMessageMap.remove(channel);

        BootNettyMqttServer.MQTTdeviceChannelGroup.remove(channel);
    }

}
